package cn.jly.flink.wordcount;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author lanyangji
 */
public class ReduceTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.fromElements(
                new Employee("Tom", 1, 1d),
                new Employee("Tom", 1, 2d),
                new Employee("Tom", 1, 3d),
                new Employee("Tom", 1, 4d),
                new Employee("Tom", 1, 5d)
        )
                .keyBy(Employee::getDeptId)
                /*
                这种方法无法求平均工资
                    6> Employee{name='Tom', deptId='1', salary=1.0}
                    6> Employee{name='Tom', deptId='1', salary=1.5}
                    6> Employee{name='Tom', deptId='1', salary=2.25}
                    6> Employee{name='Tom', deptId='1', salary=3.125}
                    6> Employee{name='Tom', deptId='1', salary=4.0625}
                 */
                .reduce(new ReduceFunction<Employee>() {
                    @Override
                    public Employee reduce(Employee employee1, Employee employee2) throws Exception {
                        employee1.setSalary((employee1.getSalary() + employee2.getSalary()) / 2);
                        return employee1;
                    }
                })
                .print();

        env.execute();
    }
}

class Employee {
    private String name;
    private Integer deptId;
    private Double salary;

    public Employee() {
    }

    public Employee(String name, Integer deptId, Double salary) {
        this.name = name;
        this.deptId = deptId;
        this.salary = salary;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public Integer getDeptId() {
        return deptId;
    }

    public void setDeptId(Integer deptId) {
        this.deptId = deptId;
    }

    public Double getSalary() {
        return salary;
    }

    public void setSalary(Double salary) {
        this.salary = salary;
    }

    @Override
    public String toString() {
        return "Employee{" +
                "name='" + name + '\'' +
                ", deptId='" + deptId + '\'' +
                ", salary=" + salary +
                '}';
    }
}